AWS Elemental MediaLive から生成されるアーカイブをいい感じに整理する仕組みを作ってみた

こんにちは、大前です。

最近は徐々に MediaLive が好きになってきた今日この頃です。

さて、今日は掲題にある通り MediaLive のアーカイブ設定に関する話となります。

MediaLive でライブ配信を行う際、アーカイブを取得する設定を入れる事が多いかと思います。

ライブ配信をしつつ、S3 にアーカイブを保存する事が出来る便利な機能です。

アーカイブをイベント毎に生成させたい

アーカイブ機能は便利ですが、同じ設定の MediaLive で何度もライブ配信を行うと同じ場所にひたすらアーカイブファイルが溜まり続けてしまい、アーカイブの管理が困難になってしまいます。

名前修飾子に $dt$ などを指定する事で一意なファイル名でアーカイブを生成する事は可能ですが、アーカイブファイルはフォルダ毎に管理できると何かと良さそうです。

• 参照:変数データの識別子

MediaLive で配信を開始する前に API 等を叩いて MediaLive のアーカイブ生成先を変更する方法なども考えられますが、なるべく自動化したいと思い、今回は EventBridge + Lambda を使って、アーカイブファイルを整理する仕組みを考えてみました。

仕組み

構成としては下記になります。

MediaLive のステータスを EventBridge で監視し、MediaLive のステータスが STOPPED（配信停止） になったタイミングで Lambda を実行します。

Lambda は日付などを用いて S3 上に新規フォルダを作成し、生成されたアーカイブファイルを整理する処理を行います。（厳密には S3 にフォルダという概念はありませんが、イメージのしやすさからフォルダという言葉を使っています）

やってみた

MediaLive を含めた、ライブ配信に必要な設定は完了しているものとします。

Lambda 関数の作成

ランタイムは Python 3.8 で Lambda 関数を作成します。

以下の関数を作成しました。

ポイントをいくつか挙げておきます。難しいことはしていないのでよしなに改修してお使いください。

• 環境変数
  • TARGET_BUCKET ... MediaLive で設定したアーカイブ保存先の S3 バケット名を指定してください
  • TARGET_BUCKET_PREFIX_[A|B] ... MediaLive の Channel Class が STANDARD の場合、パイプライン毎にアーカイブ設定をするため、S3 の保存先キーをそれぞれ設定してください
• ファイル整理処理
  • MediaLive から生成されるアーカイブファイルを別フォルダにコピー（フォルダ名は日時を使用）
  • コピー完了後、MediaLive から生成されたファイルを削除

import json
import urllib.parse
import boto3
import os
from datetime import datetime, timedelta, timezone

TARGET_BUCKET = os.environ['TARGET_BUCKET'] # アーカイブが格納されているバケット
TARGET_BUCKET_PREFIX_A = os.environ['TARGET_BUCKET_PREFIX_A'] # アーカイブファイルのキーA（フォルダ名）
TARGET_BUCKET_PREFIX_B = os.environ['TARGET_BUCKET_PREFIX_B'] # アーカイブファイルのキーB（フォルダ名）

JST = timezone(timedelta(hours=+9), 'JST')

def get_targetlist(s3, pipeline):
    target_prefix = TARGET_BUCKET_PREFIX_A if pipeline == "0" else TARGET_BUCKET_PREFIX_B
    # S3 からオブジェクト取得
    res = s3.list_objects_v2(
        Bucket = TARGET_BUCKET,
        Prefix = target_prefix
        )
    
    # 返却するリストを生成
    result_list = []
    for obj in res['Contents']:
        # list_objectsでフォルダ名のみの文字列も返却されるため弾く
        if obj['Key'].split('/')[1] != '':
            result_list.append(obj['Key'])

    return result_list


def lambda_handler(event, context):
    # 各種クライアント初期化
    s3 = boto3.client('s3')
    
    # イベントタイプ取得
    event_type = event["detail"]["state"]
    # イベントパイプライン取得
    pipeline = event["detail"]["pipeline"]
    
    try:
        if event_type == "STOPPED":
            # 移動対象となるアーカイブファイルのリスト取得
            targetlist = get_targetlist(s3, pipeline)
            if len(targetlist) == 0:
                # ファイルが見つからなかったら終了
                print('No Old Target Found')
                return

            # コピー先の prefix を生成
            dt_now = datetime.now(JST)
            copy_target_prefix = dt_now.strftime('%Y%m%d-%H%M%S') + '-' + pipeline
            
            # 過去のアーカイブファイルの移動実施
            for target in targetlist:
                file_name = target.rsplit('/', 1)[1]
                copy_to_path = os.path.join(copy_target_prefix, file_name)

                s3.copy_object(
                    Bucket = TARGET_BUCKET,
                    Key = copy_to_path,
                    CopySource = {
                        'Bucket': TARGET_BUCKET,
                        'Key': target
                    }
                )

            # 過去のアーカイブファイルを移動前ディレクトリから削除
            for target in targetlist:
                print('Delete {}'.format(target))
                s3.delete_object(
                    Bucket = TARGET_BUCKET,
                    Key = target
                )
        else:
            print('Catch Unexpected Event')
            return
    except Exception as e:
        print(e)
        return
    
    return

また、S3 への操作が発生するためアクセス権限の追加が必要です。今回は以下ポリシーを割り当てました。

{
    "Version": "2012-10-17",
    "Statement": [
        {
            "Sid": "VisualEditor0",
            "Effect": "Allow",
            "Action": [
                "s3:PutObject",
                "s3:GetObject",
                "s3:ListBucket",
                "s3:DeleteObject"
            ],
            "Resource": [
                "arn:aws:s3:::{アーカイブ保存先のS3バケット}",
                "arn:aws:s3:::{アーカイブ保存先のS3バケット}/*"
            ]
        },
        {
            "Sid": "VisualEditor1",
            "Effect": "Allow",
            "Action": "s3:ListAllMyBuckets",
            "Resource": "*"
        }
    ]
}

注意点として、生成されるアーカイブファイルの量が多いと Lambda 関数がタイムアウトする可能性もあるため、1回の配信で想定されるアーカイブファイルの量に合わせて適切なリソースとタイムアウト設定をしてください。

EventBridge の設定

続いて MediaLive のステータスをみて Lambda を起動する EventBridge ルールを作成します。

先ほど作成した Lambda 関数をターゲットに指定します。

イベントパターンは以下で作成しました。"resource" 内の ARN は作成済みの MediaLive Channel に合わせて修正してください。

"state" で STOPPED を指定する事で、MediaLive の配信停止を検知する事ができます。

{
  "source": [
    "aws.medialive"
  ],
  "resources": [
    "arn:aws:medialive:ap-northeast-1:{アカウントID}:channel:{チャネルID}"
  ],
  "detail-type": [
    "MediaLive Channel State Change"
  ],
  "detail": {
    "state": [
      "STOPPED"
    ]
  }
}

動かしてみる

実際に MediaLive の配信を行い、想定通りにアーカイブファイルが整理されるか確認してみます。

ライブ配信を開始

少し配信したら、停止

MediaLive が停止した事を確認し、S3 バケットを確認すると日付でフォルダが作成され、中にアーカイブファイルが格納されている事が確認できました。(今回は STANDARD チャネルで片方にしか入力していないため、アーカイブは 1パイプライン分のみ生成されています)

おわりに

MediaLive で生成されるアーカイブファイルを、EventBridge と Lambda 関数を使っていい感じに整理する仕組みを考えてみました。

MediaLive にてアーカイブ設定を入れる際には、生成されるアーカイブファイルの管理まで考慮して何かしらの仕組みを入れられると良いかと思います。

以上、AWS 事業本部の大前でした。

参考

• AWS Elemental MediaLiveのアーカイブ出力機能でライブ配信と同時にファイルにも映像を書き出す
• 参照:変数データの識別子